package com.whale.springboot.kafka.java.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

/**
 * 可以把poll回来的数据，按照分区 进行区分，并且开启多线程消费。
 */
public class CustomConsumerPartition {

    public static void main(String[] args) {
        // 1.创建消费者的配置对象
        Properties properties = new Properties();
        // 2.给消费者配置对象添加参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.179.131:9092,192.168.179.131:9093,192.168.179.131:9094");
        // 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 配置消费者组（组名任意起名） 必须
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "JKH20");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); //自动提交时异步提交，丢数据&&重复数据
//        properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"15000");// 开启自动提交之后，会在多少秒之后提交offset

        /**
         * earliest：从第一个消息开始取
         * latest： 从没有被提交的第一个消息开始取
         */
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        // 注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics = new ArrayList<>();
        topics.add("jkhtest002");
        kafkaConsumer.subscribe(topics);
//         消费某个主题的某个分区数据
//        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
//        topicPartitions.add(new TopicPartition("whale", 0));
//        kafkaConsumer.assign(topicPartitions);

        // 拉取数据打印
//        while (true) {

            // 设置 1s 中消费一批数据
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(3));
            System.out.println("当前组的数量：" + consumerRecords.count());
            try {

                Set<TopicPartition> partitions = consumerRecords.partitions();
                for (TopicPartition partition : partitions) {

                    /**
                     * 这里就可以按照分区 进行处理了，
                     * 甚至 你嫌慢，你可以多线程跑，每个分区 一个线程
                     *
                     * 如果手动提交offset
                     * 1，按消息进度同步提交
                     * 2，按分区粒度同步提交
                     * 3，按当前poll的批次同步提交
                     *
                     * 思考：如果在多个线程下
                     * 1，以上1，3的方式不用多线程
                     * 2，以上2的方式最容易想到多线程方式处理，有没有问题？ 没问题，因为offset是按照分区进行区分的，如果此时一个线程消费p1（成功），一个线程消费p2（报错），也没关系。不影响
                     */
                    List<ConsumerRecord<String, String>> pRecords = consumerRecords.records(partition);

                    for (ConsumerRecord<String, String> pRecord : pRecords) {
                        System.out.println("offset：" + pRecord.offset());
                        System.out.println("value：" + pRecord.value());
                        System.out.println("分区：" + pRecord.partition());
                        System.out.println("------------------------------------------");
                    }

                    //获取分区内最后一条消息的offset，只需要提交最后一个记录的offset即可
                    long poff = pRecords.get(pRecords.size() - 1).offset() + 1;
                    OffsetAndMetadata pom = new OffsetAndMetadata(poff);
                    Map<TopicPartition, OffsetAndMetadata> map = new HashMap<>();
                    map.put(partition,pom);

                    // 假装模拟，我在 分区=1的时候，业务操作报错了，我不提交offset
                    if (partition.partition() != 1) {
                        kafkaConsumer.commitSync(map);//这个是第二种，分区粒度提交offset
                    }
                }
            }
            catch (Exception e) {
                System.out.println("报错了~~~");
            }
//        }
    }
}